package com.fuwenjiang.spark.kafka;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * Kafka生产者
 */

/**
 * Kafka生产者
 */
public class KafkaProducer extends Thread {

    private String topic;

    private Producer<Integer, String> producer;
    private List<String> dataList;

    public KafkaProducer(String topic) {
        this.topic = topic;

        Properties properties = new Properties();

        properties.put("metadata.broker.list", KafkaProperties.BROKER_LIST);
        properties.put("serializer.class", "kafka.serializer.StringEncoder");
        properties.put("request.required.acks", "1");

        producer = new Producer<Integer, String>(new ProducerConfig(properties));
    }

    private void andData() {
        try {
            dataList = new ArrayList<String>();
            URL resource = KafkaProducer.class.getResource("/userInfoTestData.txt");
            String path = resource.getPath();
            File file = new File(path);
            BufferedReader reader = new BufferedReader(new FileReader(file));
            String line;
            while ((line = reader.readLine()) != null) {
                dataList.add(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public void run() {
        andData();
        int messageNo = 1;
        for (String s : dataList) {
            String message = "message_" + messageNo;
            producer.send(new KeyedMessage<Integer, String>(topic, s));
            System.err.println("Sent: " + s);
            messageNo++;
            try {
                Thread.sleep(2000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
